package com.jwater.core.net;

import java.io.BufferedInputStream;
import java.io.InputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.jwater.core.common.IpPort;
import com.jwater.core.common.S;
import com.jwater.core.net.NodeHandler.Callback;
import com.jwater.core.net.NodeHandler.ReturnCallback;

import jwater.io.netty.channel.ChannelHandlerContext;
import jwater.org.springframework.cglib.proxy.MethodProxy;

/**
 * @description:
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class SimpleNode implements Node {
    protected static Logger logger = LoggerFactory.getLogger(SimpleNode.class);
    public static final Integer timeout = 60;// 远程调用的超时时间,1分钟
    public static final Integer hearttime = 5;// 心跳时间,5秒
    private int connectInterval = 1000;// 客户端重连间隔
    private String nodeId;
    private NettyServer server;
    private ChannelHandlerContext ctx;
    private NettyClient client;
    private NodeHandler handler;
    private volatile boolean isClose = false;
    private volatile boolean isConnect = false;
    protected volatile boolean autoConnect = true;
    private volatile int connectCount = 0;
    private volatile int connectLimit = -1;
    protected AtomicInteger number = new AtomicInteger();// 包的流水号
    protected Map<Integer, SyncLock> syncLocks = new ConcurrentHashMap<Integer, SyncLock>();
    protected Map<Integer, Callback> asyncCallbacks = new ConcurrentHashMap<Integer, Callback>();
    protected Map<Integer, AsyncResult> asyncResults = new ConcurrentHashMap<Integer, AsyncResult>();
    private List<IpPort> hosts;
    private Thread heartThread;
    private Map<String, Object> attributes = new ConcurrentHashMap<>();
    private FileHandler fileHandler;

    public SimpleNode() {
    }

    public SimpleNode(ChannelHandlerContext ctx, String nodeId) {
        this.ctx = ctx;
        this.nodeId = nodeId;
    }

    public void setHandler(NodeHandler handler) {
        this.handler = handler;
    }

    public void setFileHandler(FileHandler fileHandler) {
        this.fileHandler = fileHandler;
    }

    public void setConnectInterval(int connectInterval) {
        this.connectInterval = connectInterval;
    }

    public void setAutoConnect(boolean autoConnect) {
        this.autoConnect = autoConnect;
    }

    public void setConnectLimit(int connectLimit) {
        this.connectLimit = connectLimit;
    }

    public String getNodeId() {
        return nodeId;
    }

    public boolean isConnect() {
        return isConnect;
    }

    public int getConnectCount() {
        return connectCount;
    }

    public void setConnect(boolean isConnect) {
        this.isConnect = isConnect;
    }

    public void startAsServer(int port) {
        try {
            server = new NettyServer();
            server.init();
            server.setHandler(handler, fileHandler);
            server.bind(port, false);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    public void startAsClient(String ip, int port, boolean async) {
        List<IpPort> hosts = new ArrayList<IpPort>();
        hosts.add(new IpPort(ip, port));
        startAsClient(hosts, async);
    }

    public void startAsClient(List<IpPort> hosts, boolean async) {
        try {
            this.hosts = hosts;
            client = new NettyClient();
            client.init();
            client.setNodeHandler(this, handler);
            heartThread = new Thread(() -> doHeart());
            heartThread.start();
            if (async) {
                NetExecutorUtils.execute(this::tryConnect);
            } else {
                tryConnect();
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    public boolean waitConnect(long timeout) {
        long time = 0;
        while (!isConnect) {
            try {
                TimeUnit.MILLISECONDS.sleep(10);
            } catch (InterruptedException e) {
                logger.error(e.toString(), e);
            }
            time += 10;
            if (time > timeout) {
                break;
            }
        }
        return isConnect;
    }

    private void doHeart() {
        while (!isClose) {
            try {
                if (isConnect) {
                    NodePacket packet = new NodePacket();
                    packet.setVersion(S.NET_VERSION);
                    packet.setDataType(NodePacket.DATATYPE_HEART);
                    packet.setNumber(number());
                    packet.setLength(0);
                    write(packet);
                }
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
            synchronized (this) {
                try {
                    this.wait(hearttime * 1000);
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }

    protected void tryConnect() {
        while (!isClose) {
            int index = new Random(System.currentTimeMillis()).nextInt(hosts.size());
            IpPort host = hosts.get(index);
            try {
                nodeId = host.ip + ":" + host.port;
                isConnect = client.connect(host.ip, host.port);
            } catch (Exception e) {
                logger.warn(e.toString());
            }
            connectCount++;
            if (isConnect) {
                break;
            } else {
                if (connectLimit != -1 && connectCount >= connectLimit) {
                    break;
                }
                if (connectInterval != 0) {
                    synchronized (this) {
                        try {
                            this.wait(connectInterval);
                        } catch (InterruptedException e) {
                            logger.error(e.toString(), e);
                        }
                    }
                }
            }
        }
    }

    public void doClientDiconnect() {
        if (!autoConnect) {
            return;
        }
        synchronized (this) {
            try {
                this.wait(100);
            } catch (InterruptedException e) {
                logger.error(e.toString(), e);
            }
        }
        tryConnect();
    }

    public void close() {
        try {
            isClose = true;
            if (server != null) {
                server.close();
            }

            if (client != null) {
                client.close();
            }

            if (heartThread != null) {
                synchronized (this) {
                    heartThread.interrupt();
                }
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    public void write(NodePacket packet) {
        if (ctx != null) {
            ctx.writeAndFlush(packet);
        } else if (client != null) {
            client.write(packet);
        }
    }

    protected synchronized int number() {
        int value = number.incrementAndGet();
        if (value == Integer.MAX_VALUE) {
            number.set(1);
            value = 1;
        }
        return value;
    }

    public void sendData(String dataType, byte[] data) {
        try {
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_DATA);
            packet.setNumber(number());
            DataPacket dataPacket = new DataPacket();
            dataPacket.setDataType(dataType);
            dataPacket.setData(data);
            byte[] bytes = dataPacket.serialize();
            packet.setLength(bytes.length);
            packet.setData(bytes);
            write(packet);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public void sendFile(String category, String fileName, InputStream fis) {
        try {
            int index = 0;
            SyncLock syncLock = null;
            syncLock = doSendFile(category, fileName, 0, null, FilePacket.TYPE_BEGIN);
            syncLock.latch.await();
            byte[] buff = new byte[2 * 1024 * 1024];
            BufferedInputStream bis = new BufferedInputStream(fis);
            int len = 0;
            while ((len = bis.read(buff)) > 0) {
                index++;
                byte[] temp = new byte[len];
                System.arraycopy(buff, 0, temp, 0, len);
                syncLock = doSendFile(category, fileName, index, temp, FilePacket.TYPE_DATA);
                syncLock.latch.await();
            }
            syncLock = doSendFile(category, fileName, index, null, FilePacket.TYPE_END);
            syncLock.latch.await();
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public void sendFileError(String category, String fileName, String errorMsg) {
        try {
            FilePacket filePacket = new FilePacket();
            filePacket.setType(FilePacket.TYPE_ERROR);
            filePacket.setCategory(category);
            filePacket.setFileName(fileName);
            filePacket.setErrorMsg(errorMsg);
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_FILE);
            packet.setNumber(number());
            byte[] bytes = filePacket.serialize();
            packet.setLength(bytes.length);
            packet.setData(bytes);
            SyncLock syncLock = new SyncLock();
            syncLocks.put(packet.getNumber(), syncLock);
            write(packet);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    private SyncLock doSendFile(String category, String fileName, int index, byte[] data, short type) {
        try {
            FilePacket filePacket = new FilePacket();
            filePacket.setType(type);
            filePacket.setCategory(category);
            filePacket.setFileName(fileName);
            filePacket.setIndex(index);
            filePacket.setData(data);
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_FILE);
            packet.setNumber(number());
            byte[] bytes = filePacket.serialize();
            packet.setLength(bytes.length);
            packet.setData(bytes);
            SyncLock syncLock = new SyncLock();
            syncLocks.put(packet.getNumber(), syncLock);
            write(packet);
            return syncLock;
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
        return null;
    }

    public byte[] syncCall(String method, byte[] args, long timeout) {
        try {
            return doSyncCall(method, args, timeout);
        } catch (Exception e) {
            throw new NetException(e.toString(), e);
        }
    }

    private byte[] doSyncCall(String method, byte[] args, long timeout) throws Exception {
        if (!isConnect) {
            throw new NetException("not connect");
        }
        CallPacket callPacket = new CallPacket();
        byte[] methodBytes = method.getBytes(S.charset);
        callPacket.setMethodLen(methodBytes.length);
        callPacket.setMethod(methodBytes);
        if (args != null) {
            callPacket.setArgsLen(args.length);
            callPacket.setArgs(args);
        }
        NodePacket packet = new NodePacket();
        packet.setVersion(S.NET_VERSION);
        packet.setDataType(NodePacket.DATATYPE_SYNCCALL);
        packet.setNumber(number());
        byte[] data = callPacket.serialize();
        packet.setLength(data.length);
        packet.setData(data);
        SyncLock syncLock = new SyncLock();
        syncLocks.put(packet.getNumber(), syncLock);
        write(packet);
        if (timeout == 0) {
            syncLock.latch.await();
        } else {
            syncLock.latch.await(timeout, TimeUnit.SECONDS);// 线程等待
        }

        syncLocks.remove(packet.getNumber());
        ResultPacket resultPacket = syncLock.resultPacket;
        if (resultPacket != null) {
            return resultPacket.getResult();
        } else {
            if (syncLock.exception != null) {
                throw syncLock.exception;
            } else {
                throw new NetException("execute timeout");
            }
        }
    }

    public byte[] syncCall(String method, byte[] args) {
        return syncCall(method, args, timeout);
    }

    public void asyncCall(String method, byte[] args, Callback callback) {
        if (!isConnect) {
            throw new NetException("not connect");
        }
        try {
            CallPacket callPacket = new CallPacket();
            byte[] methodBytes = method.getBytes(S.charset);
            callPacket.setMethodLen(methodBytes.length);
            callPacket.setMethod(methodBytes);
            if (args != null) {
                callPacket.setArgsLen(args.length);
                callPacket.setArgs(args);
            }
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_ASYNCCALL);
            packet.setNumber(number());
            byte[] data = callPacket.serialize();
            packet.setLength(data.length);
            packet.setData(data);
            write(packet);
            asyncCallbacks.put(packet.getNumber(), callback);
        } catch (Exception e) {
            throw new NetException(e.toString(), e);
        }
    }

    public <T> T getProxy(Class<?> clazz) {
        return getProxy(clazz, timeout);
    }

    @SuppressWarnings("unchecked")
    public <T> T getProxy(Class<?> clazz, long timeout) {
        ObjectProxy proxy = new ObjectProxy(this);
        try {
            return (T) proxy.getProxyObject(clazz.getName(), timeout);
        } catch (ClassNotFoundException e) {
            throw new NetException(e.toString(), e);
        }
    }

    protected void doHandler(NodePacket nodePacket) {
        NetExecutorUtils.execute(() -> {
            int dataType = nodePacket.getDataType();
            if (dataType == NodePacket.DATATYPE_HEART) {
                doForHeart(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_HEART_RESPONSE) {
                doForHeartResponse(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_DATA) {
                doForData(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_SYNCCALL) {
                doForSyncCall(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_ASYNCCALL) {
                doForAsyncCall(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_SYNCRESULT) {
                doForSyncResult(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_ASYNCRESULT) {
                doForAsyncResult(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_INVOKE) {
                doForInvoke(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_RETURN) {
                doForReturn(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_FILE) {
                doForFile(nodePacket);
            } else if (dataType == NodePacket.DATATYPE_FILE_RESULT) {
                doForFileResult(nodePacket);
            } else {
                throw new NetException("not support dataType:" + dataType);
            }
        });
    }

    private void doForFile(NodePacket nodePacket) {
        try {
            FilePacket filePacket = new FilePacket();
            filePacket.deserialize(nodePacket.getData());
            if (fileHandler != null) {
                if (FilePacket.TYPE_BEGIN == filePacket.getType()) {
                    fileHandler.onBegin(filePacket.getCategory(), filePacket.getFileName());
                } else if (FilePacket.TYPE_DATA == filePacket.getType()) {
                    fileHandler.onData(filePacket.getCategory(), filePacket.getFileName(), filePacket.getIndex(),
                            filePacket.getData());
                } else if (FilePacket.TYPE_END == filePacket.getType()) {
                    fileHandler.onEnd(filePacket.getCategory(), filePacket.getFileName(), filePacket.getIndex());
                } else if(FilePacket.TYPE_ERROR == filePacket.getType()){
                    fileHandler.onError(filePacket.getCategory(), filePacket.getFileName(), filePacket.getErrorMsg());
                }
            }
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_FILE_RESULT);
            packet.setNumber(nodePacket.getNumber());
            packet.setLength(0);
            write(packet);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    private void doForFileResult(NodePacket nodePacket) {
        // 直接移除包
        SyncLock syncLock = syncLocks.remove(nodePacket.getNumber());
        if (syncLock != null) {
            syncLock.latch.countDown();
        } else {
            logger.warn("file message number:" + nodePacket.getNumber() + " SyncLock is null!");
        }
    }

    protected void doForHeart(NodePacket nodePacket) {
        NodePacket packet = new NodePacket();
        packet.setVersion(S.NET_VERSION);
        packet.setDataType(NodePacket.DATATYPE_HEART_RESPONSE);
        packet.setNumber(nodePacket.getNumber());
        packet.setLength(0);
        write(packet);
    }

    protected void doForHeartResponse(NodePacket nodePacket) {

    }

    protected void doForData(NodePacket nodePacket) {
        try {
            if (handler != null) {
                DataPacket dataPacket = new DataPacket();
                dataPacket.deserialize(nodePacket.getData());
                handler.onData(this, dataPacket.getDataType(), dataPacket.getData());
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected void doForSyncCall(NodePacket nodePacket) {
        try {
            CallPacket callPacket = new CallPacket();
            callPacket = (CallPacket) callPacket.deserialize(nodePacket.getData());
            byte[] resultBytes = null;
            if (handler != null) {
                resultBytes = handler.onSyncCall(this, callPacket.getMethod(), callPacket.getArgs());
            }
            ResultPacket resultPacket = new ResultPacket();
            if (resultBytes != null) {
                resultPacket.setResultLen(resultBytes.length);
                resultPacket.setResult(resultBytes);
            }
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_SYNCRESULT);
            packet.setNumber(nodePacket.getNumber());
            byte[] data = resultPacket.serialize();
            packet.setLength(data.length);
            packet.setData(data);
            write(packet);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected void doForAsyncCall(NodePacket nodePacket) {
        try {
            CallPacket callPacket = new CallPacket();
            callPacket = (CallPacket) callPacket.deserialize(nodePacket.getData());
            if (handler != null) {
                handler.onAsyncCall(this, callPacket.getMethod(), callPacket.getArgs(),
                        new AsyncCallback(nodePacket.getNumber()));
            } else {
                new AsyncCallback(nodePacket.getNumber()).doReturn(null, null);
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected class AsyncCallback implements ReturnCallback {
        private int number;

        public AsyncCallback(int number) {
            this.number = number;
        }

        @Override
        public void doReturn(byte[] result, Exception exception) {
            try {
                ResultPacket resultPacket = new ResultPacket();
                if (result != null) {
                    resultPacket.setResultLen(result.length);
                    resultPacket.setResult(result);
                }
                NodePacket packet = new NodePacket();
                packet.setVersion(S.NET_VERSION);
                packet.setDataType(NodePacket.DATATYPE_ASYNCRESULT);
                packet.setNumber(number);
                byte[] data = resultPacket.serialize();
                packet.setLength(data.length);
                packet.setData(data);
                write(packet);
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
        }
    }

    protected void doForSyncResult(NodePacket nodePacket) {
        try {
            ResultPacket resultPacket = new ResultPacket();
            resultPacket = resultPacket.deserialize(nodePacket.getData());
            SyncLock syncLock = syncLocks.get(nodePacket.getNumber());
            if (syncLock != null) {
                syncLock.resultPacket = resultPacket;
                syncLock.latch.countDown();
            } else {
                logger.warn("message number:" + nodePacket.getNumber() + " SyncLock is null!");
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected void doForAsyncResult(NodePacket nodePacket) {
        try {
            ResultPacket resultPacket = new ResultPacket();
            resultPacket = resultPacket.deserialize(nodePacket.getData());
            Callback callback = asyncCallbacks.remove(nodePacket.getNumber());
            callback.onResult(this, resultPacket.getResult(), null);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected void doForInvoke(NodePacket nodePacket) {
        ReturnPacket returnPacket = new ReturnPacket();
        try {
            InvokePacket invokePacket = new InvokePacket();
            invokePacket = invokePacket.deserialize(nodePacket.getData());
            String methodName = invokePacket.getMethodName();
            Object[] argObjs = invokePacket.getArgObjs();
            Object result = callMethod(invokePacket.getClazzName(), methodName, argObjs);
            if (result != null) {
                returnPacket.setReturnClazzName(result.getClass().getCanonicalName());
                returnPacket.setReturnObj(result);
            }
        } catch (Throwable e) {
            logger.error(e.toString(), e);
            returnPacket.setErrorCode((short) 1);
            returnPacket.setErrorMsg(e.toString());
        }
        try {
            NodePacket packet = new NodePacket();
            packet.setVersion(S.NET_VERSION);
            packet.setDataType(NodePacket.DATATYPE_RETURN);
            packet.setNumber(nodePacket.getNumber());
            byte[] data = returnPacket.serialize();
            packet.setLength(data.length);
            packet.setData(data);
            write(packet);
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected Object callMethod(String superClazzName, String methodName, Object[] argObjs) throws Throwable {
        String clazzName = RegisterRpcService.getServiceClazzName(superClazzName);
        if (clazzName == null) {
            logger.error("callMethod clazzName is null:" + superClazzName);
            return new NetException("clazzName is null:" + superClazzName);
        }
        Object result = null;
        Object serviceInstance = RegisterRpcService.getServiceInstance(clazzName);
        Method[] methods = serviceInstance.getClass().getMethods();
        for (Method method : methods) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            if (methodName.equals(method.getName()) && parameterTypes.length == argObjs.length) {
                try {
                    result = method.invoke(serviceInstance, argObjs);
                } catch (InvocationTargetException e) {
                    throw e.getTargetException();
                }
                break;
            }
        }
        return result;
    }

    protected void doForReturn(NodePacket nodePacket) {
        try {
            ReturnPacket returnPacket = new ReturnPacket();
            returnPacket = returnPacket.deserialize(nodePacket.getData());
            AsyncResult asyncResult = asyncResults.remove(nodePacket.getNumber());
            if (asyncResult != null) {
                asyncResult.onResult(returnPacket.getReturnObj());
            } else {
                SyncLock syncLock = syncLocks.get(nodePacket.getNumber());
                if (syncLock != null) {
                    syncLock.returnPacket = returnPacket;
                    syncLock.latch.countDown();
                } else {
                    logger.warn("message number:" + nodePacket.getNumber() + " SyncLock is null");
                }
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
    }

    protected Object intercept(long timeout, String superClazz, Object obj, Method method, Object[] args,
                               MethodProxy proxy, AsyncResult asyncResult) throws Exception {
        // 屏蔽finalize方法的调用，该方法是jdk自动调用的方法
        if (method.getName().equals("finalize")) {
            return null;
        }
        if (!isConnect) {
            throw new NetException("not connect");
        }
        if (asyncResult != null) {
            interceptForAsync(superClazz, obj, method, args, proxy, asyncResult);
            return null;
        }
        InvokePacket invokePacket = new InvokePacket();
        invokePacket.setClazzName(superClazz);
        invokePacket.setMethodName(method.getName());
        invokePacket.setArgObjs(args);

        NodePacket packet = new NodePacket();
        packet.setVersion(S.NET_VERSION);
        packet.setDataType(NodePacket.DATATYPE_INVOKE);
        packet.setNumber(number());
        byte[] data = null;
        try {
            data = invokePacket.serialize();
        } catch (Exception e) {
            throw new NetException(e.toString(), e);
        }
        packet.setLength(data.length);
        packet.setData(data);
        SyncLock syncLock = new SyncLock();
        syncLocks.put(packet.getNumber(), syncLock);
        write(packet);
        try {
            if (timeout == 0) {
                syncLock.latch.await();
            } else {
                syncLock.latch.await(timeout, TimeUnit.SECONDS);// 线程等待
            }
        } catch (InterruptedException e) {
            throw new NetException(e.toString(), e);
        }
        syncLocks.remove(packet.getNumber());
        ReturnPacket returnPacket = syncLock.returnPacket;
        if (returnPacket != null) {
            if (returnPacket.getErrorCode() != 0) {
                throw new NetException(returnPacket.getErrorMsg());
            }
            return returnPacket.getReturnObj();
        } else {
            if (syncLock.exception != null) {
                throw syncLock.exception;
            } else {
                throw new NetException("execute timeout");
            }
        }
    }

    protected void interceptForAsync(String superClazz, Object obj, Method method, Object[] args, MethodProxy proxy,
                                     AsyncResult asyncResult) {
        InvokePacket invokePacket = new InvokePacket();
        invokePacket.setClazzName(superClazz);
        invokePacket.setMethodName(method.getName());
        invokePacket.setArgObjs(args);

        NodePacket packet = new NodePacket();
        packet.setVersion(S.NET_VERSION);
        packet.setDataType(NodePacket.DATATYPE_INVOKE);
        packet.setNumber(number());
        byte[] data = new byte[0];
        try {
            data = invokePacket.serialize();
        } catch (Exception e) {
            logger.error(e.toString(), e);
        }
        packet.setLength(data.length);
        packet.setData(data);
        write(packet);
        asyncResults.put(packet.getNumber(), asyncResult);
    }

    @Override
    public void setAttribute(String key, Object value) {
        this.attributes.put(key, value);
    }

    @Override
    public Object getAttribute(String key) {
        return this.attributes.get(key);
    }

    @Override
    public void removeAttribute(String key) {
        this.attributes.remove(key);
    }

    protected class SyncLock {
        protected CountDownLatch latch = new CountDownLatch(1);
        protected ResultPacket resultPacket;
        protected ReturnPacket returnPacket;
        protected Exception exception;
    }
}
